package org.atguigu.gmall.realtime.function;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.atguigu.gmall.realtime.util.*;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.function
 * Author: ZengHaiFeng
 * CreateTime:2023/5/5 18:50:32
 * Description: Todo
 * Version：1.0
 */
public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T,T> implements DimFunction<T> {
    private ThreadPoolExecutor threadPool;
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
    }

    @Override
    public void asyncInvoke(T bean, ResultFuture<T> resultFuture) {
        threadPool.submit(() -> {
            Jedis redisClient = RedisUtil.getRedisClient();
            Connection conn = DruidDSUtil.getConnection();
            JSONObject dim = DimUtil.readDim(redisClient, conn, getTable(), getId(bean));
            //System.out.println("dim complete");
            addDim(bean,dim);
            //System.out.println("add dim complete");

            redisClient.close();
            JdbcUtil.closeConnection(conn);
            //System.out.println("close cnn complete");
            //System.out.println(bean);
            resultFuture.complete(Collections.singletonList(bean));
            //System.out.println("result future complete");
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) {
        throw new RuntimeException("================ async time out ==============");
    }
}
